import redis
import time
import json
import threading


class RedisServer:
    def __init__(self, host, port, db, password):
        self.host = host
        self.port = port
        self.db = db
        self.password = password
        # 连接到Redis服务器
        self.client = redis.StrictRedis(host=self.host, port=self.port, db=self.db, password=self.password)
        self.pubsub = self.client.pubsub()
        self.messages = []

    def subscribe(self, channel):
        # 订阅给定的频道
        self.pubsub.subscribe(channel)
        # 开始监听消息
        thread = threading.Thread(target=self._listen)
        thread.daemon = True
        thread.start()

    def _listen(self):
        # 持续监听来自pubsub的消息
        for message in self.pubsub.listen():
            if message['type'] == 'message':
                data = message['data'].decode('utf-8')
                try:
                    data = json.loads(data)
                    self.messages.append(data)
                    # print(data)
                except Exception as e:
                    print("json.loads()失败，请查找原因", e)
                # 清除缓存数据
                if len(self.messages) > 1000:
                    self.messages.pop(0)

    def publish(self, channel, message):
        # 向指定频道发布指定消息
        self.client.publish(channel, message)
        # print(f"Published: {message} to {channel}")

    def get_messages(self):
        return self.messages

    def publish_by_flie(self, channel, json_path, fps=10):
        # 向指定频道发布json_path中每行的数据消息，10Hz
        with open(json_path, 'r', encoding='utf-8') as f:
            while True:
                message = f.readline()
                if not message:
                    break
                self.client.publish(channel, message)
                time.sleep((1/fps))


if __name__ == "__main__":
    redis_server = RedisServer('10.102.1.113', 6379, 0, 'Wanji@300552!')
    channel = 'ceshi'
    path = r'/home/gj/scenario_runner-0.9.13/srunner/TJ/12.json'
    threading.Thread(target=redis_server.publish_by_flie, args=(channel, path,)).start()
    threading.Thread(target=redis_server.subscribe, args=(channel,)).start()



